package belf.migrate.engine.job;

import belf.migrate.api.common.DatabaseIdentifier;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.taskconf.ConnectionConf;
import belf.migrate.api.taskconf.TaskConf;
import belf.migrate.api.util.JdbcUrlUtil;
import belf.migrate.plugin.dameng.DamengCatalog;
import belf.migrate.plugin.gbase.GBaseCatalog;
import belf.migrate.plugin.shentong.ShentongCatalog;
import belf.migrate.plugin.kingbase.KingbaseCatalog;
import belf.migrate.plugin.mysql.MySQLCatalog;
import belf.migrate.plugin.oracle.OracleCatalog;
import belf.migrate.plugin.sqlserver.SqlServerCatalog;
import lombok.extern.slf4j.Slf4j;

import java.sql.Connection;

/**
 * 为各种Job的执行提供便利的方法，创建源端和目标端的Catalog。
 */
@Slf4j
public class CatalogCreator {
    public static AbstractJdbcCatalog createSourceCatalog(TaskConf taskConf) {
        return createCatalog(taskConf.getSourceConf().getCatalogName(),
                taskConf.getSourceConf().getDatabaseName(),
                taskConf.getSourceConf().getSchemaName(),
                taskConf.getSourceConf().getConnectionConf(),
                taskConf.getSourceConf().isCaseSensitive());
    }

    public static AbstractSinkCatalog createSinkCatalog(TaskConf taskConf) {
        String catalogName = taskConf.getSinkConf().getCatalogName();
        ConnectionConf connectionConf = taskConf.getSinkConf().getConnectionConf();
        String schemaName = taskConf.getSinkConf().getSchemaName();
        boolean caseSensitive = taskConf.getSinkConf().isCaseSensitive();
        JdbcUrlUtil.UrlInfo urlInfo = JdbcUrlUtil.getUrlInfo(connectionConf.getUrl());
        switch (catalogName) {
            case DatabaseIdentifier.DAMENG:
                return new DamengCatalog(connectionConf.getUser(),
                        connectionConf.getPassword(),
                        urlInfo,
                        schemaName,
                        true);
            case DatabaseIdentifier.KINGBASE:
                return new KingbaseCatalog(catalogName,
                        connectionConf.getUser(),
                        connectionConf.getPassword(),
                        urlInfo,
                        schemaName,
                        caseSensitive);
            case DatabaseIdentifier.GBASE_8S:
                return new GBaseCatalog(
                        connectionConf.getUser(),
                        connectionConf.getPassword(),
                        urlInfo,
                        schemaName,
                        caseSensitive);
            case DatabaseIdentifier.SHENTONG:
                return new ShentongCatalog(
                        catalogName,
                        connectionConf.getUser(),
                        connectionConf.getPassword(),
                        urlInfo,
                        schemaName,
                        caseSensitive);
            default:
                break;
        }

        log.error("Fatal Error!!!No Valid Sink Catalog Found For {}!!!", catalogName);
        return null;

    }

    public static AbstractJdbcCatalog createCatalog(String catalogName,
                                                    String databaseName,
                                                    String schemaName,
                                                    ConnectionConf connectionConf,
                                                    boolean caseSensitive) {
        JdbcUrlUtil.UrlInfo urlInfo = JdbcUrlUtil.getUrlInfo(connectionConf.getUrl());
        switch (catalogName) {
            case DatabaseIdentifier.DAMENG:
                return new DamengCatalog(connectionConf.getUser(),
                        connectionConf.getPassword(),
                        urlInfo,
                        schemaName,
                        true);
            case DatabaseIdentifier.MYSQL:
                return new MySQLCatalog(connectionConf.getUser(),
                        connectionConf.getPassword(),
                        urlInfo,
                        schemaName,
                        caseSensitive);
            case DatabaseIdentifier.ORACLE:
                return new OracleCatalog(connectionConf.getUser(),
                        connectionConf.getPassword(),
                        urlInfo,
                        schemaName,
                        caseSensitive);
            case DatabaseIdentifier.KINGBASE:
                return new KingbaseCatalog(catalogName,
                        connectionConf.getUser(),
                        connectionConf.getPassword(),
                        urlInfo,
                        schemaName,
                        caseSensitive);
            case DatabaseIdentifier.SQLSERVER:
                return new SqlServerCatalog(catalogName,
                        connectionConf.getUser(),
                        connectionConf.getPassword(),
                        urlInfo,
                        schemaName,
                        caseSensitive);
            default:
                break;
        }

        log.error("Fatal Error!!!No Valid Catalog Found For {}!!!", catalogName);
        return null;

    }
}
